package com.ds.infrastructure.hyperspace.container.endpoint.entry;

import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.common.StateEnum;
import com.ds.infrastructure.hyperspace.common.TargetEndpointStream;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.gzs.GzsService;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.impl.HyperspaceTracerContext;
import com.ds.infrastructure.hyperspace.container.internal.util.AddressesUtil;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.DecoderResultProvider;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http2.Http2CodecUtil;
import io.netty.util.ReferenceCounted;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.Set;

import static com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum.*;
import static com.ds.infrastructure.hyperspace.common.StateEnum.REQUEST_NOT_ACCEPTABLE;
import static com.ds.infrastructure.hyperspace.container.internal.http.HttpCors.ALLOW_METHODS;
import static com.ds.infrastructure.hyperspace.container.internal.http.HttpCors.MAX_AGE;
import static com.ds.infrastructure.hyperspace.container.internal.http.NettyHttpHeaderNames.X_FORWARDED_FOR;
import static com.ds.infrastructure.hyperspace.container.internal.http.NettyHttpHeaderNames.X_REAL_IP;
import static io.netty.channel.ChannelFutureListener.CLOSE;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
import static io.netty.handler.codec.http.HttpHeaderNames.*;
import static io.netty.handler.codec.http.HttpHeaderValues.TEXT_PLAIN;
import static io.netty.handler.codec.http.HttpMethod.OPTIONS;
import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS;
import static io.netty.handler.codec.http.LastHttpContent.EMPTY_LAST_CONTENT;

/**
 * @author  Weishaoying
 */
@Slf4j
public class Http1EntryProtocolAdapter extends AbstractEntryProtocolAdapter {

//    private static String IGNORE_MESSAGE = "Connection reset by peer";

    private static final String HYPERSPACE_WORMHOLE_URI = "/hyperspace/wormhole";

    private static final DefaultFullHttpResponse WORMHOLE_RESPONSE;

    static {
        WORMHOLE_RESPONSE = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.EMPTY_BUFFER);
        HttpHeaders responseHttpHeaders = WORMHOLE_RESPONSE.headers();
        responseHttpHeaders.setInt(HttpHeaderNames.CONTENT_LENGTH, 0);
        responseHttpHeaders.set(HttpHeaderNames.CONTENT_TYPE, TEXT_PLAIN);
    }

    private final NettyEntryEndpointChannelInitializer channelInitializer;

    private HyperspaceDuplexFlowContext currentDuplexFlowContext;

    private boolean protocolUpgradeComplete;

    private final LinkedList<HyperspaceDuplexFlowContext> suspendRequestDuplexFlowContextLinkedList;

    private boolean isSuspendRequest = false;

    private boolean isClosedConnectionByResponseContext = false;

    public Http1EntryProtocolAdapter(NettyContainer nettyContainer,
                                     NettyEntryEndpointChannelInitializer channelInitializer) {
        super(nettyContainer, channelInitializer.getEntryNodeConfig());
        this.channelInitializer = channelInitializer;
        this.suspendRequestDuplexFlowContextLinkedList = new LinkedList<>();
    }

    @Override
    public void handleChannelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof DecoderResultProvider) {
            DecoderResult decoderResult = ((DecoderResultProvider) msg).decoderResult();
            if (decoderResult.isFailure()) {
                log.warn("Receive a unsupport http request! entry {}:{}, message [{}]",
                        entryNodeConfig.getHost(), entryNodeConfig.getPort(), decoderResult.toString());
                currentDuplexFlowContext = null;
                releaseData(msg);
                closeEntryChannel();
                return;
            }
        }

        if (msg instanceof HttpRequest) {
            currentDuplexFlowContext = initDuplexFlowContext(ctx, (HttpRequest) msg);
        }
        if (currentDuplexFlowContext == null) {
            releaseData(msg);
            return;
        }
        if (!currentDuplexFlowContext.getEndpointProtocol().isLongLive()) {
            currentDuplexFlowContext.appendReferData(msg);
        }
        if (isSuspendRequest) {
            if (msg instanceof LastHttpContent) {
                handlerSuspendRequest();
                currentDuplexFlowContext = null;
            }
            return;
        }
        entryFlowOperation.handleInput(hyperspaceConnector, currentDuplexFlowContext, msg);
        if (currentDuplexFlowContext == null || currentDuplexFlowContext.isClosed()) {
            releaseData(msg);
            currentDuplexFlowContext = null;
            return;
        }
        if (msg instanceof LastHttpContent) {
            currentDuplexFlowContext.setInputComplete(true);
            entryFlowOperation.handleInputComplete(hyperspaceConnector, currentDuplexFlowContext);
            if (!currentDuplexFlowContext.getEndpointProtocol().isLongLive()) {
                if (!currentDuplexFlowContext.isClosed()) {
                    isSuspendRequest = true;
                }
                currentDuplexFlowContext = null;
            }
        }
    }

    private void handlerSuspendRequest() {
        int maxPendingRequestSize = entryNodeConfig.getProtocolConfig().getMaxEndpointPendingRequestSize();
        if (nettyContainer.getEntryEndpointPendingSize(entryNodeConfig.getPort()) > maxPendingRequestSize) {
            entryFlowOperation.interruptAndCloseContext(currentDuplexFlowContext, REQUEST_NOT_ACCEPTABLE,
                    "Entry endpoint (port:" + entryNodeConfig.getPort() + ") ,exceed max pending request size " + maxPendingRequestSize);
            return;
        }
        int maxSuspendRequestSize = entryNodeConfig.getProtocolConfig().getMaxConnectionSuspendRequestSize();
        if (suspendRequestDuplexFlowContextLinkedList.size() > maxSuspendRequestSize) {
            closeEntryChannel();
            LinkedList<Object> referDataLinkList = currentDuplexFlowContext.getReferDataLinkList();
            if (!CollectionUtils.isEmpty(referDataLinkList)) {
                for (Object referData : referDataLinkList) {
                    releaseData(referData);
                }
            }
            log.warn("Request was interrupted by {}, exceed http1.1 pipeline max suspend request size:{}, entry [{}:{}], uri [{}], headers {}",
                    StateEnum.REQUEST_BAD, maxSuspendRequestSize,
                    entryNodeConfig.getHost(), entryNodeConfig.getPort(), currentDuplexFlowContext.getUri(),
                    currentDuplexFlowContext.getRequestStartState() == null ? "null" : ((HttpRequest) currentDuplexFlowContext.getRequestStartState()).headers());
            return;
        }
        currentDuplexFlowContext.setInputComplete(true);
        suspendRequestDuplexFlowContextLinkedList.offer(currentDuplexFlowContext);
        nettyContainer.getMetricsManager().incrementEntryRequestPendingGauge(String.valueOf(entryNodeConfig.getPort()));
        nettyContainer.incrementEntryEndpointPendingSize(entryNodeConfig.getPort());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (currentDuplexFlowContext != null && !isSuspendRequest) {
            entryFlowOperation.handleInputComplete(hyperspaceConnector, currentDuplexFlowContext);
        }
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        while (!CollectionUtils.isEmpty(suspendRequestDuplexFlowContextLinkedList)) {
            HyperspaceDuplexFlowContext tempSuspendDuplexFlowContext = suspendRequestDuplexFlowContextLinkedList.poll();
            nettyContainer.getMetricsManager().decrementEntryRequestPendingGauge(String.valueOf(entryNodeConfig.getPort()));
            nettyContainer.decrementEntryEndpointPendingSize(entryNodeConfig.getPort());
            LinkedList<Object> referDataLinkList = tempSuspendDuplexFlowContext.getReferDataLinkList();
            if (!CollectionUtils.isEmpty(referDataLinkList)) {
                for (Object referData : referDataLinkList) {
                    releaseData(referData);
                }
            }
            log.warn("Request was interrupted by {}, release suspend request because connection is closed , entry [{}:{}], uri [{}], headers {}",
                    StateEnum.REQUEST_BAD, entryNodeConfig.getHost(), entryNodeConfig.getPort(), tempSuspendDuplexFlowContext.getUri(),
                    tempSuspendDuplexFlowContext.getRequestStartState() == null ? "null" : ((HttpRequest) tempSuspendDuplexFlowContext.getRequestStartState()).headers());
        }
        super.handlerRemoved(ctx);
    }

    @Override
    public HyperspaceDuplexFlowContext currentDuplexFlowContext() {
        return currentDuplexFlowContext;
    }

    @Override
    public void writeToEntry(HyperspaceDuplexFlowContext duplexFlowContext, Object msg) {
        if (!channel.isActive()) {
            releaseData(msg);
            return;
        }
        if (msg instanceof HttpResponse) {
            HttpResponse httpResponse = (HttpResponse) msg;
            if (!HttpUtil.isKeepAlive(httpResponse)) {
                isClosedConnectionByResponseContext = true;
            }
        }
        if (msg instanceof LastHttpContent) {
            if (duplexFlowContext.getEndpointProtocol().isLongLive()) {
                HttpResponse httpResponse = (HttpResponse) duplexFlowContext.getTargetResponseState();
                if (httpResponse != null && SWITCHING_PROTOCOLS.equals(httpResponse.status())) {
                    channelHandlerContext.writeAndFlush(msg).addListener(
                            new ProtocolUpgradeFutureListener(channelInitializer,
                                    entryFlowOperation, nettyContainer.getGzsService(), duplexFlowContext));
                    return;
                }
            } else {
                channelHandlerContext.writeAndFlush(msg).addListener(isClosedConnectionByResponseContext ?
                        CLOSE : new HttpCloseFutureListener(entryFlowOperation, duplexFlowContext));
                return;
            }
        }
        channelHandlerContext.write(msg).addListener(CLOSE_ON_FAILURE);
    }

//    @Override
//    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//        if (cause instanceof IOException && IGNORE_MESSAGE.equals(cause.getMessage())) {
//            return;
//        }
//        super.exceptionCaught(ctx, cause);
//    }

    @Override
    protected void resetChannelCodecState(HyperspaceDuplexFlowContext duplexFlowContext) {
        if (duplexFlowContext.getEndpointProtocol().getIndex() < WEBSOCKET.getIndex()) {
            channelHandlerContext.write(EMPTY_LAST_CONTENT);
        }
    }

    @Override
    protected void interruptToEntryComplete(HyperspaceDuplexFlowContext duplexFlowContext) {
        StateEnum stateEnum = duplexFlowContext.getInterruptState();
        log.warn("Request was interrupted by {}, {}, entry [{}:{}], uri [{}], headers {}",
                stateEnum.name(), duplexFlowContext.getInterruptMessage(),
                entryNodeConfig.getHost(), entryNodeConfig.getPort(), duplexFlowContext.getUri(),
                duplexFlowContext.getRequestStartState() == null ? "null" :
                        ((HttpRequest) duplexFlowContext.getRequestStartState()).headers());
        if (!channel.isActive()) {
            return;
        }
        if (!protocolUpgradeComplete || duplexFlowContext.getEndpointProtocol().getIndex() < WEBSOCKET.getIndex()) {
            DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.valueOf(stateEnum.getHttpCode(), stateEnum.getMessage()), false);
            HttpHeaders httpHeaders = resp.headers();
            httpHeaders.setInt(HttpHeaderNames.CONTENT_LENGTH, 0);
            httpHeaders.set(HttpHeaderNames.CONTENT_TYPE, TEXT_PLAIN);
            channelHandlerContext.writeAndFlush(resp).addListener(
                    new HttpCloseFutureListener(entryFlowOperation, duplexFlowContext));
            return;
        }
        closeEntryChannel();
    }

    @Override
    public void releaseData(Object data) {
        if (data instanceof ReferenceCounted) {
            ReferenceCounted referenceCounted = (ReferenceCounted) data;
            if (referenceCounted.refCnt() > 0) {
                referenceCounted.release();
            }
        }
    }

    @Override
    public void clearSuspendRequest() {
        if (CollectionUtils.isEmpty(suspendRequestDuplexFlowContextLinkedList)) {
            isSuspendRequest = false;
            return;
        }
        if (!channel.isActive()) {
            return;
        }
        HyperspaceDuplexFlowContext suspendDuplexFlowContext = suspendRequestDuplexFlowContextLinkedList.poll();
        if (suspendDuplexFlowContext != null) {
            nettyContainer.getMetricsManager().decrementEntryRequestPendingGauge(String.valueOf(entryNodeConfig.getPort()));
            nettyContainer.decrementEntryEndpointPendingSize(entryNodeConfig.getPort());
            LinkedList<Object> referDataLinkList = suspendDuplexFlowContext.getReferDataLinkList();
            if (!CollectionUtils.isEmpty(referDataLinkList)) {
                for (Object referData : referDataLinkList) {
                    entryFlowOperation.handleInput(hyperspaceConnector, suspendDuplexFlowContext, referData);
                }
            }
            if (suspendDuplexFlowContext.checkTimeout()) {
                suspendDuplexFlowContext.onTimeout();
                return;
            }
            entryFlowOperation.handleInputComplete(hyperspaceConnector, suspendDuplexFlowContext);
        }
    }

    @Override
    public void releaseDuplexFlowContext() {

    }

    @Override
    public String getEntryHeader(HyperspaceDuplexFlowContext duplexFlowContext, String headerName) {
        if (duplexFlowContext == null) {
            return null;
        }
        HttpRequest httpRequest = (HttpRequest) duplexFlowContext.getRequestStartState();
        return httpRequest.headers().get(headerName);
    }

    @Override
    public byte[] getEntryBody(HyperspaceDuplexFlowContext duplexFlowContext) {
        if (duplexFlowContext == null) {
            return null;
        }
        return ByteBufUtil.getBytes(((FullHttpRequest) duplexFlowContext.getRequestStartState()).content());
    }

    private HyperspaceDuplexFlowContext initDuplexFlowContext(ChannelHandlerContext ctx, HttpRequest httpRequest) {
        HttpHeaders httpHeaders = httpRequest.headers();

        if (httpRequest.uri().startsWith(HYPERSPACE_WORMHOLE_URI)) {
            boolean keepAlive = HttpUtil.isKeepAlive(httpRequest);
            HttpUtil.setKeepAlive(WORMHOLE_RESPONSE, keepAlive);
            ChannelFuture future = ctx.writeAndFlush(WORMHOLE_RESPONSE);
            if (!keepAlive) {
                future.addListener(CLOSE);
            }
            return null;
        }

        if (httpRequest.method().equals(OPTIONS) && httpHeaders.contains(HttpHeaderNames.ORIGIN) &&
                httpHeaders.contains(ACCESS_CONTROL_REQUEST_METHOD)) {
            handlePreflightRequest(ctx, httpRequest, httpHeaders);
            return null;
        }

        EndpointProtocolEnum endpointProtocol = HTTP;
        if (httpHeaders.containsValue(HttpHeaderNames.CONNECTION, HttpHeaderValues.UPGRADE, true)) {
            if (httpHeaders.contains(HttpHeaderNames.UPGRADE, HttpHeaderValues.WEBSOCKET, true)) {
                endpointProtocol = WEBSOCKET;
            } else if (httpHeaders.contains(HttpHeaderNames.UPGRADE,
                    Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME, true)) {
                endpointProtocol = HTTP2;
            }
        }

        HyperspaceDuplexFlowContext duplexFlowContext = new HyperspaceDuplexFlowContext(
                nettyContainer, this, httpRequest,
                httpHeaders.get(X_FORWARDED_FOR), httpHeaders.get(X_REAL_IP),
                clientIp, endpointProtocol, null);
        duplexFlowContext.setHost(httpHeaders.get(HttpHeaderNames.HOST));
        duplexFlowContext.setUri(httpRequest.uri());
        duplexFlowContext.setRequestType(httpRequest.method().name());
        if (nettyContainer.getOpentracingConfig().isEnable()) {
            duplexFlowContext.setTracerContext(
                    new HyperspaceTracerContext(AddressesUtil.getIpAndPort(channel.localAddress()), clientIp, httpRequest));
            duplexFlowContext.getTracerContext()
                    .sample(duplexFlowContext.getRequestStartTime(), nettyContainer.getOpentracingConfig());
        }
        return duplexFlowContext;
    }

    private void handlePreflightRequest(ChannelHandlerContext ctx, HttpRequest httpRequest, HttpHeaders requestHeaders) {
        HttpResponse response = new DefaultFullHttpResponse(httpRequest.protocolVersion(),
                HttpResponseStatus.OK, false);
        HttpHeaders responseHeader = response.headers();

        String requestOriginStr = requestHeaders.get(HttpHeaderNames.ORIGIN);
        if (StringUtil.isNullOrEmpty(requestOriginStr)) {
            responseHeader.set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, "*");
        } else {
            responseHeader.set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, requestOriginStr);
            responseHeader.set(HttpHeaderNames.VARY, HttpHeaderNames.ORIGIN);
            responseHeader.set(ACCESS_CONTROL_ALLOW_CREDENTIALS, "true");
        }
        String requestMethodStr = requestHeaders.get(ACCESS_CONTROL_REQUEST_METHOD);
        Set<HttpMethod> allowMethod;
        if (StringUtil.isNullOrEmpty(requestMethodStr)) {
            allowMethod = ALLOW_METHODS;
        } else {
            allowMethod = new HashSet<>();
            allowMethod.add(HttpMethod.valueOf(requestMethodStr));
        }
        responseHeader.set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS, allowMethod);
        String requestAllowHeaderStr = requestHeaders.get(ACCESS_CONTROL_REQUEST_HEADERS);
        if (!StringUtil.isNullOrEmpty(requestAllowHeaderStr)) {
            responseHeader.set(ACCESS_CONTROL_ALLOW_HEADERS, requestAllowHeaderStr);
        }
        responseHeader.set(ACCESS_CONTROL_MAX_AGE, MAX_AGE);
        if (!responseHeader.contains(CONTENT_LENGTH)) {
            responseHeader.set(CONTENT_LENGTH, HttpHeaderValues.ZERO);
        }
        boolean keepAlive = HttpUtil.isKeepAlive(httpRequest);
        HttpUtil.setKeepAlive(response, keepAlive);
        ChannelFuture future = ctx.writeAndFlush(response);
        if (!keepAlive) {
            future.addListener(CLOSE);
        }
    }

    private class HttpCloseFutureListener implements GenericFutureListener<ChannelFuture> {

        private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

        private final HyperspaceDuplexFlowContext duplexFlowContext;

        private HttpCloseFutureListener(EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation,
                                        HyperspaceDuplexFlowContext duplexFlowContext) {
            this.entryFlowOperation = entryFlowOperation;
            this.duplexFlowContext = duplexFlowContext;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                Throwable cause = future.cause();
                log.error("Occur an exception that entry endpoint handler unexpected!", cause);
                entryFlowOperation.handleExceptionCaught(duplexFlowContext, cause);
                closeEntryChannel();
                return;
            }
            HttpRequest httpRequest = (HttpRequest) duplexFlowContext.getRequestStartState();
            if (!HttpUtil.isKeepAlive(httpRequest)) {
                closeEntryChannel();
            }
        }
    }

    private class ProtocolUpgradeFutureListener implements GenericFutureListener<ChannelFuture> {

        private final NettyEntryEndpointChannelInitializer channelInitializer;

        private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

        private final GzsService gzsService;

        private final HyperspaceDuplexFlowContext duplexFlowContext;

        private ProtocolUpgradeFutureListener(NettyEntryEndpointChannelInitializer channelInitializer,
                                              EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation,
                                              GzsService gzsService, HyperspaceDuplexFlowContext duplexFlowContext) {
            this.channelInitializer = channelInitializer;
            this.entryFlowOperation = entryFlowOperation;
            this.gzsService = gzsService;
            this.duplexFlowContext = duplexFlowContext;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                Throwable cause = future.cause();
                log.error("Occur an exception while writing protocol switch response!", cause);
                entryFlowOperation.handleExceptionCaught(duplexFlowContext, cause);
                closeEntryChannel();
                return;
            }
            HttpResponse httpResponse = (HttpResponse) duplexFlowContext.getTargetResponseState();
            HttpHeaders httpResponseHeaders = httpResponse.headers();
            EndpointProtocolEnum switchToProtocol;
            if (httpResponseHeaders.containsValue(HttpHeaderNames.UPGRADE,
                    Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME, true)) {
                switchToProtocol = HTTP2;
            } else if (httpResponseHeaders.contains(HttpHeaderNames.UPGRADE,
                    HttpHeaderValues.WEBSOCKET, true)) {
                switchToProtocol = WEBSOCKET;
            } else {
                channel.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                        HttpResponseStatus.UPGRADE_REQUIRED, Unpooled.EMPTY_BUFFER, false))
                        .addListener(CLOSE);
                return;
            }
            channelInitializer.switchProtocolFromHttp(channel.pipeline(),
                    duplexFlowContext, switchToProtocol);
            TargetEndpointStream targetEndpointStream = duplexFlowContext.getCurrentReferTargetEndpointStream();
            if (targetEndpointStream != null) {
                targetEndpointStream.switchProtocol(duplexFlowContext, switchToProtocol);
            }
            if (gzsService != null) {
                gzsService.register(duplexFlowContext);
            }
            protocolUpgradeComplete = true;
        }
    }
}
